package com.qst.mapreduce.wordcount.join;


import com.qst.mapreduce.wordcount.diliuzhang.dierjie.App;
import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IOUtils;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;


import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.net.URI;
import java.util.HashMap;

public class JoinAppMapper extends Mapper<LongWritable, Text, Text, App> {
    //增加一个全局的变量  存在一个映射关系  存的是对应的id和对应的value
    HashMap<String, String> nameMap=new HashMap();

    @Override
    protected void setup(Mapper<LongWritable, Text, Text, App>.Context context) throws IOException, InterruptedException {
        //通过缓存文件得到姓名文件地址
        URI[] cacheFile=context.getCacheFiles();
        Path path=new Path(cacheFile[0]);
        //获取文件系统对象，并开流
        FileSystem fs= FileSystem.get(context.getConfiguration());
        FSDataInputStream fis=fs.open(path);
        //通过包装流转换为reader，方便按行读取
        BufferedReader reader=new BufferedReader(new InputStreamReader(fis, "UTF-8"));
        //逐行读取，按行处理
        String line;
        while(StringUtils.isNotEmpty(line= reader.readLine())){
            String[] split=line.split("\t");
            nameMap.put(split[0],split[1]);

        }
        //关闭流
        IOUtils.closeStream(reader);
    }

    @Override
    protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, App>.Context context) throws IOException, InterruptedException {
        String line=value.toString();
        // /t 制表符 分割
        String[] words=line.split(",");

        System.out.println(nameMap);
        String id=words[1];
        String name=nameMap.get(id);
        App App=new App();

        //Text outkey=new Text(words[2]);
        App.setCount(Integer.parseInt(words[3]));
        App.setTime(Integer.parseInt(words[4]));
        App.setConsume(Integer.parseInt(words[5]));
        App.setName(name);

        context.write(new Text(words[2]), App);



    }
}